package it.java.giw.dataextraction.aleph.wrapper.filmit;

import it.java.giw.dataextraction.aleph.helper.CSVHelper;

import java.io.File;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingQueue;

public class WrapperMaster {
	private LinkedBlockingQueue<String> datasetFiles;
	private int numworkers;
	private ExecutorService workerPool;

	public WrapperMaster(int numworkers, LinkedBlockingQueue<String> datasetFiles) {
		this.datasetFiles = datasetFiles;
		this.numworkers = numworkers;
		this.workerPool = Executors.newFixedThreadPool(numworkers);
	}

	public void start() {
		CountDownLatch startSignal = new CountDownLatch(1);
		CountDownLatch doneSignal = new CountDownLatch(numworkers);
		
		List<LinkedBlockingQueue<String>> chunks = createChunks(datasetFiles);
		CSVHelper helper = new CSVHelper();
		for(int i=0; i<numworkers;i++){
			//richiedo all'executor service di avviare un task per ogni worker thread disponibile
			LinkedBlockingQueue<String> chunk = new LinkedBlockingQueue<String>(chunks.get(i));
			workerPool.execute(new FilmitWrapperWorker(chunk, startSignal, doneSignal, helper));
		}
		startSignal.countDown();// let all threads proceed
		try {
			doneSignal.await();// wait for all to finish
			workerPool.shutdownNow();
			
		} catch (InterruptedException e) {
			workerPool.shutdownNow();
		}
	}

	private List<LinkedBlockingQueue<String>> createChunks(LinkedBlockingQueue<String> datasetFiles) {
		
		List<LinkedBlockingQueue<String>> output = new ArrayList<LinkedBlockingQueue<String>>();
		int filesPerChunk = datasetFiles.size()/numworkers;
		for(int i=0; i<numworkers; i++){
			int j=0;
			LinkedBlockingQueue<String> chunk = new LinkedBlockingQueue<String>();
			while(j<filesPerChunk){
				chunk.add(datasetFiles.poll());
				j++;
			}
			output.add(chunk);
		}
		while(!datasetFiles.isEmpty()){
			output.get(0).add(datasetFiles.poll());
		}
		return output;
	}

	private static LinkedBlockingQueue<String> listFilesForFolder(final File folder) {
		LinkedBlockingQueue<String> output = new LinkedBlockingQueue<String>();
		for (final File fileEntry : folder.listFiles()) {
			if (fileEntry.isDirectory()) {
				listFilesForFolder(fileEntry);
			} else {
				output.add(fileEntry.getName());
			}
		}
		return output;
	}

	public static void main(String args[]) throws Exception{

		final File folder = new File("download");
		LinkedBlockingQueue<String> datasetFiles = new LinkedBlockingQueue<String>();
		datasetFiles.addAll(listFilesForFolder(folder));
		int numWorkers = Runtime.getRuntime().availableProcessors();
		
		String[] info = new String[10];
		info[1] = "TITLE";
		info[2] = "DIRECTORS";
		info[3] = "GENRE";
		info[4] = "YEAR";
		info[5] = "RATING";
		info[6] = "ACTORS";
		info[7] = "POSTER";
		info[8] = "DIRECTOR BIOGRAPHY";
		
		CSVHelper helper = new CSVHelper();
		helper.writeCSV(info,"Collection FilmIt.csv");

		WrapperMaster mywrapper = new WrapperMaster(numWorkers, datasetFiles);
		mywrapper.start();
	}

}
